package com.consumer;

import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.PredictionIoService;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ServiceManager;
import com.producer.PiwikLogActionDataScanner;
import com.producer.PiwikLogLinkVisitActionDataScanner;
import com.producer.PiwikLogVisitDataScanner;

@Configuration
@ConfigurationProperties(prefix="kafka")
public class ConsumerConfig  {
	@Autowired
	PredictionIoService predictionIoService;
	private  String servers;
 
	 public String getServers() {
		return servers;
	}

	public void setServers(String servers) {
		this.servers = servers;
	}
	
	  @Bean(name="piwikConsumerServiceManager")
	  public ServiceManager piwikConsumerServiceManager(){
		  List<Consumer> list = Lists.newArrayList();
		  list.add(getConsumer(PiwikLogActionDataScanner.topic));
		  list.add(getConsumer(PiwikLogLinkVisitActionDataScanner.topic));
		  list.add(getConsumer(PiwikLogVisitDataScanner.topic));
		  return new ServiceManager(list);
	  }
	  
	  Consumer getConsumer(String topic){
			Consumer consumer = new Consumer(topic,servers);			
			consumer.setPredictionIoService(predictionIoService);
			return consumer;
		}
	

}
